package com.sparrow.actor;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Objects;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * MailBox
 * 1同 executor 任务复用当前线程连续执行。
 * 2不同 executor 任务交给目标 executor 执行
 */
public class MailBox {

    private static final Logger logger = LoggerFactory.getLogger(MailBox.class);

    private final Queue<Envelope> queue = new ConcurrentLinkedQueue<>();

    private final AtomicBoolean running = new AtomicBoolean(false);

    /**
     * 入队任务并尝试执行
     */
    public void enqueue(Runnable task, Executor executor) {
        Objects.requireNonNull(task, "Task must not be null");
        Objects.requireNonNull(executor, "Executor must not be null");

        queue.offer(new Envelope(task, executor));
        if (running.compareAndSet(false, true)) {
            Envelope first = queue.poll();
            if (first != null) {
                first.executor.execute(first);
            }
        }
    }

    private class Envelope implements Runnable {
        Runnable task;
        final Executor executor;

        Envelope(Runnable task, Executor executor) {
            this.task = task;
            this.executor = executor;
        }

        @Override
        public void run() {
            while (task != null) {
                try {
                    task.run();
                } catch (Exception e) {
                    logger.error("MailBox task exec error", e);
                }
                this.task = null;

                Envelope next = queue.poll();
                if (next == null) {
                    // 队列为空，尝试安全释放 running
                    if (running.compareAndSet(true, false)) {
                        // 再次检查，防止新任务入队时漏掉
                        if (!queue.isEmpty() && running.compareAndSet(false, true)) {
                            continue;
                        }
                    }
                    break;
                }
                // 同 executor，直接在当前线程执行
                if (executor == next.executor) {
                    this.task = next.task;
                } else {
                    next.executor.execute(next);
                }
            }
        }
    }
}
